//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// 
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Lesser General Public License for more details.
// 
// You should have received a copy of the GNU Lesser General Public License
// along with this program.  If not, see http://www.gnu.org/licenses/.
// 

#include "Cloud.h"
#include "BTStatistics.h"
#include "BitTorrentMsg_m.h"
#include "InternalMsg_m.h"
#include <cassert>


namespace peermaliziosi {

using namespace std;

Define_Module(Cloud);

#define DISABLE_DATASTREAM

Cloud::Cloud() {

}

Cloud::~Cloud() {
}

void Cloud::initialize() {
    numPeers= gateSize("io");

    outgoingMsg= new ExtractOutgoingMsg*[numPeers];
    for (int k=0; k<numPeers; k++) {
        outgoingMsg[k]= NULL;
    }

    outgoing= new list<BitTorrentMsg*>[numPeers];


    for (int k=0; k<numPeers; k++) {
        //string str= outgoingStr(k);
        string str= " (contenuto della coda omesso)";
        EV << "outgoing[" << k << "] (size " << outgoing[k].size() << "): " << str << endl;
    }

    // statistica
    totMsgOnFlySignal= registerSignal("totMsgOnFly");
    totByteOnFlySignal= registerSignal("totByteOnFly");
    totRequestSignal= registerSignal("totRequest");
    totPieceSignal= registerSignal("totPiece");
    totCancelSignal= registerSignal("totCancel");

}

void Cloud::finish() {
    for (int k=0; k<numPeers; k++) {
        cancelAndDelete(outgoingMsg[k]);
    }
    delete[] outgoingMsg;

    // pulisco le code
    for (int k=0; k<numPeers; k++) {
        while (!outgoing[k].empty()) {
            delete outgoing[k].front();
            outgoing[k].pop_front();
        }
    }
    delete[] outgoing;
}

void Cloud::handleMessage(cMessage *msg) {

    // anche se va in crash, su stdout si ha la stampa del numero di eventi
    //cout << "> Cloud, evento " << simulation.getEventNumber() <<", msgName " << msg->getName() << "\n";
    //cout.flush();

    if(BitTorrentMsg *tmsg = dynamic_cast<BitTorrentMsg *>(msg)){

        pushOutgoingMsg(tmsg);

    } else if(ExtractOutgoingMsg *xmsg = dynamic_cast<ExtractOutgoingMsg *>(msg)) {
        // estrazione di msg dalla coda
        popOutgoingMsgAndSend(xmsg->getQueueIndex());
        delete xmsg;

    } else {
        EV <<"Messaggio " <<msg->getName() <<" non gestito" << endl;;
        throw std::runtime_error("Inviato alla nuvola un messaggio non gestito");
    }
}

void Cloud::pushOutgoingMsg(BitTorrentMsg *msg) {

    if ( (dynamic_cast<Request *>(msg)!=NULL) ||
            (dynamic_cast<Cancel *>(msg)!=NULL) ||
            (dynamic_cast<Piece *>(msg)!=NULL) ) {
        // solo in questi casi faccio statistica
        emit(totMsgOnFlySignal, BTStatistics::addMsgOnFly());
        emit(totByteOnFlySignal, BTStatistics::addByteOnFly(msg->getBtPayloadLength()));

        if(dynamic_cast<Request *>(msg)!=NULL){

            emit(totRequestSignal, BTStatistics::addRequestCloud());

        }

        else if (dynamic_cast<Piece *>(msg)!=NULL){

            emit(totPieceSignal, BTStatistics::addPieceCloud());
        }

        else{ // Cancel
            emit(totCancelSignal, BTStatistics::addCancelCloud());

        }


    }

    // inoltro al destinatario
    int dest= msg->getDestination();

    cChannel *txChannel = gate("io$o", dest)->getTransmissionChannel();

    if (outgoingMsg[dest]==NULL) {
        // invio al destinatario il msg di inizio stream
#ifndef DISABLE_DATASTREAM
        DataStream *dmsg= new DataStream("DataStream");
        dmsg->setStart(true);
        EV << "Invio al peer " << dest << " il messaggio di inizio data-stream" << endl;
        send(dmsg, "io$o", dest);
#endif
    }

    simtime_t txFinishTime = txChannel->getTransmissionFinishTime(); // tempo al quale posso inviare
    EV << "TransmissionFinishTime: " << txFinishTime << ", now: " << simTime() << endl;
    if ( outgoingMsg[dest]==NULL ) {
        EV << "Inoltro (senza passare dalla coda) \"" << msg->getName() << "\" al peer " << dest << endl;;
        // TODO assert rimovibili
        assert(txFinishTime <= simTime()); // canale libero
        assert(outgoing[dest].empty()); // come e' possibile che il canale sia libero ma esistono pkt in coda di uscita? non puo' essere
        assert(!txChannel->isBusy()); // canale libero
        // channel free; send out packet immediately
        send(msg, "io$o", dest); // invio

        if ( (dynamic_cast<Request *>(msg)!=NULL) ||
                (dynamic_cast<Cancel *>(msg)!=NULL) ||
                (dynamic_cast<Piece *>(msg)!=NULL) ) {
            // solo in questi casi faccio statistica
            emit(totMsgOnFlySignal, BTStatistics::delMsgOnFly());
            emit(totByteOnFlySignal, BTStatistics::delByteOnFly(msg->getBtPayloadLength()));

            if(dynamic_cast<Request *>(msg)!=NULL){

                emit(totRequestSignal, BTStatistics::delRequestCloud());

            }

            else if (dynamic_cast<Piece *>(msg)!=NULL){

                emit(totPieceSignal, BTStatistics::delPieceCloud());
            }

            else{ // Cancel
                emit(totCancelSignal, BTStatistics::delCancelCloud());

            }

        }
    } else {
        // store packet and schedule timer; when the timer expires,
        // the packet should be removed from the queue and sent out

        EV << "Aggiungo \"" << msg->getName() << "\" alla coda del peer " << dest << endl;
        outgoing[dest].push_back(msg);

        //string str= outgoingStr(dest);
        string str= " (contenuto della coda omesso)";
        EV << "outgoing[" << dest << "] (size " << outgoing[dest].size() << "): " << str << endl;
    }

    if (outgoingMsg[dest]==NULL) {
        // non ci sono messaggi ExtractOutgoingMsg schedulati => creo un nuovo msg
        EV << "Schedulo un nuovo messaggio di estrazione dalla coda del peer " << dest << endl;
        outgoingMsg[dest]= new ExtractOutgoingMsg("ExtractOutgoingMsg");
        outgoingMsg[dest]->setQueueIndex(dest);
        txFinishTime = txChannel->getTransmissionFinishTime();
        scheduleAt(txFinishTime, outgoingMsg[dest]);
    }
}


void Cloud::popOutgoingMsgAndSend(int queueIndex) {

    if (outgoing[queueIndex].empty()) {
        // non ci sono piu' pacchetti da inviare: fine dello stream
        outgoingMsg[queueIndex]= NULL;

#ifndef DISABLE_DATASTREAM
        DataStream *dmsg= new DataStream("DataStream");
        dmsg->setStart(false);
        EV << "Invio al peer " << queueIndex << " il messaggio di fine data-stream" << endl;
        send(dmsg, "io$o", queueIndex);
#endif
        return;
    }

    // estrazione dalla coda

    BitTorrentMsg *msg= outgoing[queueIndex].front();
    if ( (dynamic_cast<Request *>(msg)!=NULL) ||
            (dynamic_cast<Cancel *>(msg)!=NULL) ||
            (dynamic_cast<Piece *>(msg)!=NULL) ) {
        // solo in questi casi faccio statistica
        emit(totMsgOnFlySignal, BTStatistics::delMsgOnFly());
        emit(totByteOnFlySignal, BTStatistics::delByteOnFly(msg->getBtPayloadLength()));

        if(dynamic_cast<Request *>(msg)!=NULL){

            emit(totRequestSignal, BTStatistics::delRequestCloud());

        }

        else if (dynamic_cast<Piece *>(msg)!=NULL){

            emit(totPieceSignal, BTStatistics::delPieceCloud());
        }

        else{ // Cancel
            emit(totCancelSignal, BTStatistics::delCancelCloud());

        }

    }

    int dest= msg->getDestination();

    EV << "Estraggo dalla coda ed inoltro \"" << msg->getName() << "\" al peer " << dest << endl;
    send(msg, "io$o", dest); // invio
    outgoing[queueIndex].pop_front();

    //string str= outgoingStr(dest);
    string str= " (contenuto della coda omesso)";
    EV << "outgoing[" << dest << "] (size " << outgoing[dest].size() << "): " << str << endl;

    // in ogni caso schedulo un nuovo msg di estrazione
    cChannel *txChannel = gate("io$o", dest)->getTransmissionChannel();
    simtime_t txFinishTime = txChannel->getTransmissionFinishTime(); // tempo al quale posso inviare
    EV << "Schedulo un nuovo messaggio di estrazione dalla coda del peer " << queueIndex << endl;
    outgoingMsg[queueIndex]= new ExtractOutgoingMsg("ExtractOutgoingMsg");
    outgoingMsg[queueIndex]->setQueueIndex(queueIndex);
    scheduleAt(txFinishTime, outgoingMsg[queueIndex]);
}

string& Cloud::outgoingStr(int queueIndex) {
    string *str= new string();
    *str= "[";
    for (list<BitTorrentMsg*>::iterator it=outgoing[queueIndex].begin(); it!=outgoing[queueIndex].end();) {
        *str= *str + "\"" + (*it)->getName() + "\"";
        ++it;
        if ( it != outgoing[queueIndex].end()) {
            *str= *str + ",";
        }
    }
    *str= *str + "]";
    return *str;
}

} //namespace
